package com.young.springboot.kafka;

import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.boot.context.properties.PropertyMapper;
import org.springframework.boot.convert.DurationUnit;
import org.springframework.kafka.listener.ContainerProperties;
import org.springframework.util.CollectionUtils;
import org.springframework.util.unit.DataSize;

import java.time.Duration;
import java.time.temporal.ChronoUnit;
import java.util.*;

/**
 * @description：
 * @author: yangyahui01
 * @date: 12/5/24 2:28 PM
 */
public class KafkaProperties {

    private List<String> bootstrapServers = new ArrayList<>(Collections.singletonList("localhost:9092"));

    private String clientId;

    private final Map<String, String> properties = new HashMap<>();

    private final Template template = new Template();

    public void setBootstrapServers(List<String> bootstrapServers) {
        this.bootstrapServers = bootstrapServers;
    }

    public List<String> getBootstrapServers() {
        return bootstrapServers;
    }

    public void setClientId(String clientId) {
        this.clientId = clientId;
    }

    public String getClientId() {
        return clientId;
    }

    public Map<String, String> getProperties() {
        return properties;
    }

    public Template getTemplate() {
        return template;
    }

    /*配置属性*/
    protected Map<String, Object> buildCommonProperties() {
        Map<String, Object> properties = new HashMap<>();
        if (this.bootstrapServers != null) {
            properties.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, this.bootstrapServers);
        }
        if (this.clientId != null) {
            properties.put(CommonClientConfigs.CLIENT_ID_CONFIG, this.clientId);
        }
        if (!CollectionUtils.isEmpty(this.properties)) {
            properties.putAll(this.properties);
        }

        return properties;
    }


    public static class Consumer {

        /**
         * 如果定义了enable.auto.commit 为true,则该值代表：消费者偏移量向 kafka 提交的频率
         */
        private Duration autoCommitInterval;

        /**
         * 当 kafka 中没有初始偏移量或当前偏移量在服务器中不存在，该如何处理？
         * earliest：自动重置偏移量到最早的偏移量
         * latest：自动重置偏移量为最新的偏移量
         * none：如果消费者组原来的偏移量不存在，则向消费者抛异常
         * anything：向消费者抛异常
         */
        private String autoOffsetReset;

        /**
         * kafka 集群列表
         */
        private List<String> bootstrapServers;

        private String clientId;

        /**
         * 消费者自动周期性的向服务器提交偏移量
         */
        private Boolean enableAutoCommit;

        /**
         * 如果没有从服务器获取到一批数据的最小字节数,满足该时间了，仍然返回数据
         */
        private Duration fetchMaxWait;

        /**
         * 消费者获取服务器端一批消息最小的字节数
         */
        private DataSize fetchMinSize;

        /**
         * 标记消费者所属的消费者组
         */
        private String groupId;

        /**
         * 消费者和coordinator之间的心跳时间
         */
        private Duration heartbeatInterval;

        /**
         * 接收消息的 key 的反序列化类型
         */
        private Class<?> keyDeserializer = StringDeserializer.class;

        /**
         * 接收消息的 value 的反序列化类型
         */
        private Class<?> valueDeserializer = StringDeserializer.class;

        /**
         * 一次poll 拉取数据返回消息的最大条数
         */
        private Integer maxPollRecords;

        private final Map<String, String> properties = new HashMap<>();

        public Duration getAutoCommitInterval() {
            return this.autoCommitInterval;
        }

        public void setAutoCommitInterval(Duration autoCommitInterval) {
            this.autoCommitInterval = autoCommitInterval;
        }

        public String getAutoOffsetReset() {
            return this.autoOffsetReset;
        }

        public void setAutoOffsetReset(String autoOffsetReset) {
            this.autoOffsetReset = autoOffsetReset;
        }

        public List<String> getBootstrapServers() {
            return this.bootstrapServers;
        }

        public void setBootstrapServers(List<String> bootstrapServers) {
            this.bootstrapServers = bootstrapServers;
        }

        public String getClientId() {
            return this.clientId;
        }

        public void setClientId(String clientId) {
            this.clientId = clientId;
        }

        public Boolean getEnableAutoCommit() {
            return this.enableAutoCommit;
        }

        public void setEnableAutoCommit(Boolean enableAutoCommit) {
            this.enableAutoCommit = enableAutoCommit;
        }

        public Duration getFetchMaxWait() {
            return this.fetchMaxWait;
        }

        public void setFetchMaxWait(Duration fetchMaxWait) {
            this.fetchMaxWait = fetchMaxWait;
        }

        public DataSize getFetchMinSize() {
            return this.fetchMinSize;
        }

        public void setFetchMinSize(DataSize fetchMinSize) {
            this.fetchMinSize = fetchMinSize;
        }

        public String getGroupId() {
            return this.groupId;
        }

        public void setGroupId(String groupId) {
            this.groupId = groupId;
        }

        public Duration getHeartbeatInterval() {
            return this.heartbeatInterval;
        }

        public void setHeartbeatInterval(Duration heartbeatInterval) {
            this.heartbeatInterval = heartbeatInterval;
        }

        public Class<?> getKeyDeserializer() {
            return this.keyDeserializer;
        }

        public void setKeyDeserializer(Class<?> keyDeserializer) {
            this.keyDeserializer = keyDeserializer;
        }

        public Class<?> getValueDeserializer() {
            return this.valueDeserializer;
        }

        public void setValueDeserializer(Class<?> valueDeserializer) {
            this.valueDeserializer = valueDeserializer;
        }

        public Integer getMaxPollRecords() {
            return this.maxPollRecords;
        }

        public void setMaxPollRecords(Integer maxPollRecords) {
            this.maxPollRecords = maxPollRecords;
        }

        public Map<String, String> getProperties() {
            return this.properties;
        }

        public Map<String, Object> buildProperties() {
            Properties properties = new Properties();
            PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull();
            map.from(this::getAutoCommitInterval).asInt(Duration::toMillis)
                    .to(properties.in(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG));
            map.from(this::getAutoOffsetReset)
                    .to(properties.in(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG));
            map.from(this::getBootstrapServers)
                    .to(properties.in(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG));
            map.from(this::getClientId)
                    .to(properties.in(ConsumerConfig.CLIENT_ID_CONFIG));
            map.from(this::getEnableAutoCommit)
                    .to(properties.in(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG));
            map.from(this::getFetchMaxWait).asInt(Duration::toMillis)
                    .to(properties.in(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG));
            map.from(this::getFetchMinSize).asInt(DataSize::toBytes)
                    .to(properties.in(ConsumerConfig.FETCH_MIN_BYTES_CONFIG));
            map.from(this::getGroupId).to(properties.in(ConsumerConfig.GROUP_ID_CONFIG));
            map.from(this::getHeartbeatInterval).asInt(Duration::toMillis)
                    .to(properties.in(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG));
            map.from(this::getKeyDeserializer)
                    .to(properties.in(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG));
            map.from(this::getValueDeserializer)
                    .to(properties.in(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG));
            map.from(this::getMaxPollRecords)
                    .to(properties.in(ConsumerConfig.MAX_POLL_RECORDS_CONFIG));
            return properties.with(this.properties);
        }

    }

    public static class Producer {

        /**
         * 0：生产者发送过来的数据，不需要等数据落盘应答
         * 1：生产者发送过来的数据，Leader 收到数据后应答
         * -1（all）：生产者发送过来的数据，Leader+和 isr 队列里面的所有节点收齐数据后应答,默认值是-1
         */
        private String acks;

        /**
         * 缓冲区一批数据最大值，默认 16k。适当增加该值，可以提高吞吐量，但是如果该值设置太大，会导致数据传输延迟增加
         */
        private DataSize batchSize;

        /**
         * 生产者连接集群所需的 broker 地址
         */
        private List<String> bootstrapServers;

        /**
         * RecordAccumulator 缓冲区总大小，默认 32m
         */
        private DataSize bufferMemory;

        private String clientId;

        /**
         * 生产者发送的所有数据的压缩方式,默认是 none，也就是不压缩
         * 支持压缩类型：none、gzip、snappy、lz4 和 zstd
         */
        private String compressionType;

        /**
         * 指定发送消息的 key 和 value 的序列化类型。一定要写全类名
         */
        private Class<?> keySerializer = StringSerializer.class;

        private Class<?> valueSerializer = StringSerializer.class;

        /**
         * 当消息发送出现错误的时候，系统会重发消息
         * retries 表示重试次数。默认是 int 最大值，2147483647
         * 如果设置了重试，还想保证消息的有序性，需要设置MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION=1否则在重试此失败消息的时候，其他的消息可能发送成功
         */
        private Integer retries;

        private String transactionIdPrefix;

        private final Map<String, String> properties = new HashMap<>();

        public String getAcks() {
            return this.acks;
        }

        public void setAcks(String acks) {
            this.acks = acks;
        }

        public DataSize getBatchSize() {
            return this.batchSize;
        }

        public void setBatchSize(DataSize batchSize) {
            this.batchSize = batchSize;
        }

        public List<String> getBootstrapServers() {
            return this.bootstrapServers;
        }

        public void setBootstrapServers(List<String> bootstrapServers) {
            this.bootstrapServers = bootstrapServers;
        }

        public DataSize getBufferMemory() {
            return this.bufferMemory;
        }

        public void setBufferMemory(DataSize bufferMemory) {
            this.bufferMemory = bufferMemory;
        }

        public String getClientId() {
            return this.clientId;
        }

        public void setClientId(String clientId) {
            this.clientId = clientId;
        }

        public String getCompressionType() {
            return this.compressionType;
        }

        public void setCompressionType(String compressionType) {
            this.compressionType = compressionType;
        }

        public Class<?> getKeySerializer() {
            return this.keySerializer;
        }

        public void setKeySerializer(Class<?> keySerializer) {
            this.keySerializer = keySerializer;
        }

        public Class<?> getValueSerializer() {
            return this.valueSerializer;
        }

        public void setValueSerializer(Class<?> valueSerializer) {
            this.valueSerializer = valueSerializer;
        }

        public Integer getRetries() {
            return this.retries;
        }

        public void setRetries(Integer retries) {
            this.retries = retries;
        }

        public String getTransactionIdPrefix() {
            return this.transactionIdPrefix;
        }

        public void setTransactionIdPrefix(String transactionIdPrefix) {
            this.transactionIdPrefix = transactionIdPrefix;
        }

        public Map<String, String> getProperties() {
            return this.properties;
        }

        public Map<String, Object> buildProperties() {
            Properties properties = new Properties();
            PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull();
            map.from(this::getAcks).to(properties.in(ProducerConfig.ACKS_CONFIG));
            map.from(this::getBatchSize).asInt(DataSize::toBytes)
                    .to(properties.in(ProducerConfig.BATCH_SIZE_CONFIG));
            map.from(this::getBootstrapServers)
                    .to(properties.in(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG));
            map.from(this::getBufferMemory).as(DataSize::toBytes)
                    .to(properties.in(ProducerConfig.BUFFER_MEMORY_CONFIG));
            map.from(this::getClientId)
                    .to(properties.in(ProducerConfig.CLIENT_ID_CONFIG));
            map.from(this::getCompressionType)
                    .to(properties.in(ProducerConfig.COMPRESSION_TYPE_CONFIG));
            map.from(this::getKeySerializer)
                    .to(properties.in(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG));
            map.from(this::getRetries).to(properties.in(ProducerConfig.RETRIES_CONFIG));
            map.from(this::getValueSerializer)
                    .to(properties.in(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG));
            return properties.with(this.properties);
        }

    }

    public static class Admin {

        private String clientId;

        private final Map<String, String> properties = new HashMap<>();

        private boolean failFast;

        public String getClientId() {
            return this.clientId;
        }

        public void setClientId(String clientId) {
            this.clientId = clientId;
        }

        public boolean isFailFast() {
            return this.failFast;
        }

        public void setFailFast(boolean failFast) {
            this.failFast = failFast;
        }

        public Map<String, String> getProperties() {
            return this.properties;
        }

        public Map<String, Object> buildProperties() {
            Properties properties = new Properties();
            PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull();
            map.from(this::getClientId)
                    .to(properties.in(ProducerConfig.CLIENT_ID_CONFIG));
            return properties.with(this.properties);
        }

    }


    public static class Listener {

        public enum Type {

            SINGLE,

            BATCH

        }

        private Listener.Type type = Listener.Type.SINGLE;

        private ContainerProperties.AckMode ackMode;

        private String clientId;

        private Integer concurrency;

        private Duration pollTimeout;

        private Float noPollThreshold;

        private Integer ackCount;

        private Duration ackTime;

        private Duration idleEventInterval;

        @DurationUnit(ChronoUnit.SECONDS)
        private Duration monitorInterval;

        private Boolean logContainerConfig;

        public Listener.Type getType() {
            return this.type;
        }

        public void setType(Listener.Type type) {
            this.type = type;
        }

        public ContainerProperties.AckMode getAckMode() {
            return this.ackMode;
        }

        public void setAckMode(ContainerProperties.AckMode ackMode) {
            this.ackMode = ackMode;
        }

        public String getClientId() {
            return this.clientId;
        }

        public void setClientId(String clientId) {
            this.clientId = clientId;
        }

        public Integer getConcurrency() {
            return this.concurrency;
        }

        public void setConcurrency(Integer concurrency) {
            this.concurrency = concurrency;
        }

        public Duration getPollTimeout() {
            return this.pollTimeout;
        }

        public void setPollTimeout(Duration pollTimeout) {
            this.pollTimeout = pollTimeout;
        }

        public Float getNoPollThreshold() {
            return this.noPollThreshold;
        }

        public void setNoPollThreshold(Float noPollThreshold) {
            this.noPollThreshold = noPollThreshold;
        }

        public Integer getAckCount() {
            return this.ackCount;
        }

        public void setAckCount(Integer ackCount) {
            this.ackCount = ackCount;
        }

        public Duration getAckTime() {
            return this.ackTime;
        }

        public void setAckTime(Duration ackTime) {
            this.ackTime = ackTime;
        }

        public Duration getIdleEventInterval() {
            return this.idleEventInterval;
        }

        public void setIdleEventInterval(Duration idleEventInterval) {
            this.idleEventInterval = idleEventInterval;
        }

        public Duration getMonitorInterval() {
            return this.monitorInterval;
        }

        public void setMonitorInterval(Duration monitorInterval) {
            this.monitorInterval = monitorInterval;
        }

        public Boolean getLogContainerConfig() {
            return this.logContainerConfig;
        }

        public void setLogContainerConfig(Boolean logContainerConfig) {
            this.logContainerConfig = logContainerConfig;
        }

    }

    public static class Template {

        private String defaultTopic;

        public String getDefaultTopic() {
            return this.defaultTopic;
        }

        public void setDefaultTopic(String defaultTopic) {
            this.defaultTopic = defaultTopic;
        }

    }

    private static class Properties extends HashMap<String, Object> {

        public <V> java.util.function.Consumer<V> in(String key) {
            return (value) -> put(key, value);
        }

        public Properties with( Map<String, String> properties) {
            putAll(properties);
            return this;
        }

    }
}
